"""
数据源服务模块
处理数据源的连接、订阅等操作
"""

import asyncio
import json
from asyncua import Client as OPCUAClient
import paho.mqtt.client as mqtt
from django.conf import settings
from functools import partial
import threading
import queue
import logging
import time

# 初始化日志记录器
logger = logging.getLogger(__name__)

class MQTTClient:
    """MQTT客户端封装类
    
    提供异步接口和回调处理
    """

    def __init__(self, host, port, username=None, password=None):
        """初始化MQTT客户端
        
        Args:
            host: MQTT服务器地址
            port: MQTT服务器端口
            username: 用户名（可选）
            password: 密码（可选）
        """
        self.client = mqtt.Client()
        if username:
            self.client.username_pw_set(username, password)

        logger.info("初始化MQTT客户端")
        self.client.on_connect = self._on_connect
        self.client.on_message = self._on_message
        self.client.on_disconnect = self._on_disconnect

        self.host = host
        self.port = port
        self.subscriptions = {}  # 存储主题订阅的回调函数
        self._connected = threading.Event()  # 使用threading.Event替代asyncio.Event
        self._reconnect = True  # 控制是否需要重连
        self._loop = None  # 存储事件循环引用

    def _on_connect(self, client, userdata, flags, rc):
        """MQTT连接回调"""
        if rc == 0:
            logger.info(f"MQTT连接成功 - Host: {self.host}, Port: {self.port}")
            self._connected.set()  # 直接设置连接标志
            # 重新订阅之前的主题
            for topic in self.subscriptions:
                client.subscribe(topic)
                logger.info(f"重新订阅主题: {topic}")
        else:
            logger.error(f"MQTT连接失败 - Host: {self.host}, Port: {self.port}, 错误代码: {rc}")
            self._connected.clear()

    def _on_message(self, client, userdata, msg):
        """MQTT消息回调"""
        if msg.topic in self.subscriptions:
            # 解码消息内容
            try:
                payload_str = msg.payload.decode('utf-8')
            except UnicodeDecodeError:
                payload_str = msg.payload.decode('gbk')

            logger.info(f"收到MQTT消息 - Topic: {msg.topic}, Payload: {payload_str}")
            callback = self.subscriptions.get(msg.topic)
            if callback and self._loop:
                # 在事件循环中执行回调
                asyncio.run_coroutine_threadsafe(callback(msg.topic, payload_str), self._loop)

    def _on_disconnect(self, client, userdata, rc):
        """MQTT断开连接回调"""
        self._connected.clear()
        if rc != 0 and self._reconnect:
            logger.warning(f"MQTT意外断开连接 - Host: {self.host}, Port: {self.port}, 错误代码: {rc}")
            # 在新线程中尝试重连
            threading.Thread(target=self._reconnect_thread).start()
        else:
            logger.info(f"MQTT已断开连接 - Host: {self.host}, Port: {self.port}")

    def _reconnect_thread(self):
        """重连线程"""
        while not self._connected.is_set() and self._reconnect:
            try:
                logger.info(f"尝试重新连接MQTT - Host: {self.host}, Port: {self.port}")
                self.client.reconnect()
                if self._connected.wait(timeout=5.0):  # 等待连接成功
                    logger.info(f"MQTT重连成功 - Host: {self.host}, Port: {self.port}")
                    break
            except Exception as e:
                logger.error(f"MQTT重连失败: {str(e)}", exc_info=True)
                time.sleep(5)  # 等待5秒后重试

    async def connect(self):
        """连接到MQTT服务器"""
        # 保存当前事件循环
        self._loop = asyncio.get_event_loop()
        self._reconnect = True

        # 在新线程中启动MQTT客户端
        def _run_mqtt_client():
            try:
                self.client.connect(self.host, self.port)
                self.client.loop_forever()
            except Exception as e:
                logger.error(f"MQTT客户端运行出错: {str(e)}", exc_info=True)

        self.client_thread = threading.Thread(target=_run_mqtt_client)
        self.client_thread.daemon = True
        self.client_thread.start()

        # 等待连接成功
        if not self._connected.wait(timeout=10.0):
            raise ConnectionError("MQTT连接超时")
        logger.info(f"MQTT连接已就绪 - Host: {self.host}, Port: {self.port}")

    async def disconnect(self):
        """断开MQTT连接"""
        self._reconnect = False  # 禁用重连
        self.client.loop_stop()
        self.client.disconnect()
        self._connected.clear()
        if hasattr(self, 'client_thread'):
            self.client_thread.join(timeout=5.0)
        logger.info(f"MQTT连接已关闭 - Host: {self.host}, Port: {self.port}")

    async def subscribe(self, topic, callback):
        """订阅MQTT主题
        
        Args:
            topic: 要订阅的主题
            callback: 接收消息的回调函数
        """
        self.subscriptions[topic] = callback
        self.client.subscribe(topic)
        logger.info(f"已订阅MQTT主题: {topic}")

    async def unsubscribe(self, topic):
        """取消订阅MQTT主题"""
        if topic in self.subscriptions:
            del self.subscriptions[topic]
            self.client.unsubscribe(topic)
            logger.info(f"已取消订阅MQTT主题: {topic}")


class DataSourceService:
    """数据源服务类
    
    提供：
    1. 数据源连接测试
    2. OPC UA连接和订阅
    3. MQTT连接和订阅
    """

    _opcua_clients = {}  # 存储OPC UA客户端连接
    _mqtt_clients = {}  # 存储MQTT客户端连接

    @classmethod
    async def connect_opcua(cls, connection_string):
        """连接到OPC UA服务器
        
        Args:
            connection_string: OPC UA连接字符串
            
        Returns:
            OPCUAClient: OPC UA客户端实例
        """
        if connection_string in cls._opcua_clients:
            return cls._opcua_clients[connection_string]

        try:
            client = OPCUAClient(connection_string)
            await client.connect()
            cls._opcua_clients[connection_string] = client
            return client
        except Exception as e:
            raise ConnectionError(f"连接OPC UA服务器失败: {str(e)}")

    @classmethod
    async def connect_mqtt(cls, connection_string):
        """连接到MQTT服务器
        
        Args:
            connection_string: MQTT连接字符串
            
        Returns:
            MQTTClient: MQTT客户端实例
        """
        if connection_string in cls._mqtt_clients:
            return cls._mqtt_clients[connection_string]

        try:
            # 解析连接字符串
            # 格式：mqtt://username:password@host:port
            import re
            pattern = r'mqtt://(?:([^:@]+)(?::([^@]+))?@)?([^:]+)(?::(\d+))?'
            match = re.match(pattern, connection_string)
            if not match:
                raise ValueError("无效的MQTT连接字符串")

            username, password, host, port = match.groups()
            port = int(port) if port else 1883

            # 创建MQTT客户端
            client = MQTTClient(host, port, username, password)
            await client.connect()

            cls._mqtt_clients[connection_string] = client
            return client
        except Exception as e:
            raise ConnectionError(f"连接MQTT服务器失败: {str(e)}")

    @classmethod
    def test_connection(cls, type, connection_string):
        """测试数据源连接
        
        Args:
            type: 数据源类型 ('opcua' 或 'mqtt')
            connection_string: 连接字符串
            
        Returns:
            tuple: (是否成功, 消息)
        """
        try:
            if type == 'opcua':
                # 创建临时连接测试
                async def test_opcua():
                    client = OPCUAClient(connection_string)
                    await client.connect()
                    await client.disconnect()

                asyncio.run(test_opcua())
                return True, "OPC UA服务器连接成功"

            elif type == 'mqtt':
                # 解析连接字符串
                import re
                pattern = r'mqtt://(?:([^:@]+)(?::([^@]+))?@)?([^:]+)(?::(\d+))?'
                match = re.match(pattern, connection_string)
                if not match:
                    return False, "无效的MQTT连接字符串"

                username, password, host, port = match.groups()
                port = int(port) if port else 1883

                # 创建临时客户端测试连接
                client = mqtt.Client()
                if username:
                    client.username_pw_set(username, password)

                client.connect(host, port, 5)
                client.disconnect()
                return True, "MQTT服务器连接成功"

            else:
                return False, "不支持的数据源类型"

        except Exception as e:
            return False, f"连接失败: {str(e)}"
